package cn.me.implement.threadpool.v1_2;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * <h3>线程池 1.2 版本</h3>
 * 它实现的功能如下：
 * <ul>
 *     <li>使用一个 {@link Set} 存放所有线程</li>
 *     <li>让线程在一段时间内没有获取任务的情况下停止</li>
 *     <li>在新建线程时交给线程一个任务</li>
 *     <li>在 判断线程池大小 和 将线程放入线程池 时使用锁来保证原子性</li>
 *     <li>提供给使用者一个构造方法，指定线程的数量</li>
 *     <li>线程在提交任务时才创建</li>
 *     <li>{@link #submit(Runnable task)} 通过任务队列提交任务，线程通过任务队列领取任务</li>
 *     <li>任务队列使用 {@link ArrayBlockingQueue} 阻塞队列实现</li>
 * </ul>
 */
public class ThreadPool1_2 {

    /**
     * 线程池中线程的数量
     */
    private final int poolSize;

    /**
     * 存放线程的集合，使用 {@link Set} 是因为 {@link Set#remove(Object)} 性能更高
     */
    private final Set<Worker> threadPool = new HashSet<>();

    /**
     * 线程池的管程
     * <p>
     * 用于保证 <strong>获取线程池大小</strong>、<strong>将线程放入线程池</strong>、<strong>从线程池中移除线程</strong> 的互斥性
     */
    private final Object threadPoolMonitor = new Object();

    /**
     * 任务队列，大小为 10
     */
    private final BlockingQueue<Runnable> taskQueue = new ArrayBlockingQueue<>(10);

    /**
     * 线程池中的线程执行的任务，在指定时间内无法获取到任务时结束
     */
    private final class Worker implements Runnable {

        /**
         * 线程执行的初始任务
         */
        private Runnable initialTask;

        /**
         * 对 <strong>真正运行的线程</strong> 的引用，用于调用其 {@link Thread#start()} 方法启动线程
         */
        private final Thread actuallyRunningThread;

        public Worker(Runnable initialTask) {
            this.initialTask = initialTask;
            this.actuallyRunningThread = new Thread(this);
            threadPool.add(this);
        }

        @Override
        public void run() {
            initialTask.run();
            initialTask = null; // help GC
            while (true) {
                Runnable t = null;
                try {
                    // 取出任务，这步操作是阻塞的，如果线程没有取到任务，则会在这里阻塞 3s
                    t = taskQueue.poll(3, TimeUnit.SECONDS);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                // 检查是否获取到任务了，如果没有则退出循环，停止运行
                if (t == null) {
                    break;
                }
                t.run();
            }
            // 从线程池中移除当前线程
            synchronized (threadPoolMonitor) {
                threadPool.remove(this);
            }
        }

        /**
         * 启动内部保存的线程
         */
        public void start() {
            actuallyRunningThread.start();
        }
    }

    /**
     * 构造一个线程池
     *
     * @param poolSize 线程池中的线程数量
     */
    public ThreadPool1_2(int poolSize) {
        this.poolSize = poolSize;
    }

    /**
     * 提交任务
     *
     * @param task 待执行的任务
     */
    public void submit(Runnable task) {
        synchronized (threadPoolMonitor) {
            // 如果线程数量小于最大线程数量，则新建一个线程执行任务，然后直接返回
            if (threadPool.size() < poolSize) {
                Worker worker = new Worker(task);
                worker.start();
                return;
            }
        }

        // 否则将任务放到任务队列中，等待空闲线程执行
        taskQueue.offer(task);
    }

    /**
     * 获取当前线程池中的线程数量
     *
     * @return 当前线程池中的线程数量
     */
    public int getCurrPoolSize() {
        synchronized (threadPoolMonitor) {
            return threadPool.size();
        }
    }
}
